package org.example.reactive;

import org.reactivestreams.Subscription;
import reactor.core.Disposable;
import reactor.core.publisher.BaseSubscriber;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.SignalType;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;

import java.io.IOException;
import java.time.Duration;

public class FlexDemo {

    public static void main(String[] args) throws Exception {
        thread1(args);
        System.in.read();
    }

    public static void Mono(String[] args) {
        Mono<Integer> mono = Mono.just(1);
        mono.subscribe(System.out::println);
    }

    public static void emptyFlux(String[] args) {
        Flux<Object> emptyFlux = Flux.empty().doOnComplete(() -> {
            System.out.println("流结束了。。。");
        });
        emptyFlux.subscribe(System.out::println);
    }

    public static void flux(String[] args) throws IOException, InterruptedException {
        Flux<Integer> flux = Flux.just(1, 2, 3, 4, 5);
        flux.subscribe(e -> System.out.println("e1=" + e));
        // 一个数据流可以有很多消费者
        flux.subscribe(e -> System.out.println("e2=" + e));
        //对每个消费者来说流都是一样的；广播模式
        Flux<Long> longFlux = Flux.interval(Duration.ofSeconds(1));
        longFlux.subscribe(System.out::println);
    }

    public static void flux2(String[] args) throws InterruptedException {
        Flux<Integer> range = Flux.range(2, 5)
                .delayElements(Duration.ofSeconds(1))
                .doOnComplete(() -> System.out.println("流正常结束"))
                .doOnCancel(() -> System.out.println("流已被取消"))
                .doOnError(throwable -> System.out.println("流出错" + throwable))
                .doOnNext(integer -> System.out.println("doOnNext..." + integer));
        range.subscribe(System.out::println);
        Thread.sleep(10000);
    }

    public static void flux3(String[] args) throws InterruptedException {
        Flux.range(1, 8)
//                .log()
                .filter(i -> i > 3)
                .log()
                .map(i -> "hh" + i)
//                .log()
                .subscribe(System.out::println);
        Thread.sleep(5000);
    }

    public static void flux4(String[] args) throws InterruptedException {
        Flux<Integer> range = Flux.range(2, 5)
                .doOnNext(integer -> {
                    if (integer == 5) {
//                        throw new IllegalArgumentException("error");
                    }
                });
        range.subscribe(System.out::println, throwable -> System.out.println(throwable.getMessage()), () -> System.out.println("流结束了"));
        Thread.sleep(10000);
    }

    public static void flux5(String[] args) throws InterruptedException {
        Flux<Integer> range = Flux.range(2, 5)
                .doOnNext(integer -> {
                    if (integer == 5) {
//                        throw new IllegalArgumentException("error");
                    }
                });
        // 自定义消费者
        range.subscribe(new BaseSubscriber<Integer>() {
            @Override
            protected void hookOnSubscribe(Subscription subscription) {
                super.hookOnSubscribe(subscription);
            }

            @Override
            protected void hookOnNext(Integer value) {
                super.hookOnNext(value);
                System.out.println("value = " + value);
            }

            @Override
            protected void hookOnComplete() {
                super.hookOnComplete();
            }

            @Override
            protected void hookOnError(Throwable throwable) {
                super.hookOnError(throwable);
            }

            @Override
            protected void hookOnCancel() {
                super.hookOnCancel();
            }

            @Override
            protected void hookFinally(SignalType type) {
                super.hookFinally(type);
            }
        });
        Thread.sleep(10000);
    }

    public static void buffer(String[] args) throws IOException {
        Flux.range(1, 10).buffer(3).subscribe(v -> System.out.println("v = " + v));
        /* buffer是凑够一批发一次，之前是一个发一次
        request(1)是请求buffer大小的数据
         */

    }

    public static void limit(String[] args) {
        Flux.range(1, 1000)
                .log()
                .limitRate(100)
                .subscribe();
        /*
        75%预取策略：limitRate(100)
        第一次抓取100个数据，如果75%的元素已经处理了，继续抓取新的75%元素放入缓冲区
         */
    }

    // 单线程使用，一部使用create
    public static void generate(String[] args) {
        Flux.generate(() -> 0, (state, sink) -> {
            // 0-10
            if (state > 10) {
                sink.complete();
            }
            sink.next(state);
            return state + 1;
        }).log().subscribe(System.out::println);
    }

    public static void dispose(String[] args) throws IOException {
        Disposable subscribe = Flux.interval(Duration.ofSeconds(1))
                .log()
                .subscribe(v -> System.out.println("v = " + v));

        new Thread(() -> {
            try {
                Thread.sleep(10000);
                subscribe.dispose();
            } catch (InterruptedException e) {

            }
        }).start();
    }

    public static void handle(String[] args) {
        /*
        handle可以使流中有不同类型的数据
        map只能返回一种类型数据
         */
        Flux.range(1, 10)
                .handle((value, sink) -> {
                    System.out.println("value = " + value);
                    sink.next("aaa-" + value);
                }).log().subscribe();

    }

    public static void thread(String[] args) {
        /* 响应式编程：全异步、消息、事件回调
        默认还是使用当前线程
         */

        // 调度器：线程池
//        Schedulers.immediate(); // 默认：无执行上下文，使用当前线程运行所有操作
//        Schedulers.single(); // 使用固定的单线程
//        Schedulers.boundedElastic(); // 有界、弹性调度；不是无限扩充的线程池；默认有10*CPU核心个线程；队列默认100k；默认60s
//        Schedulers.fromExecutor(new ThreadPoolExecutor(4, 8, 60, TimeUnit.SECONDS, new LinkedBlockingDeque<>(1000))); // 自定义线程池
//        Schedulers.parallel(); // 并发池，forkjoin

        Flux.range(1, 10)
                .publishOn(Schedulers.parallel()) // 改变发布者所在线程池
                .log()
                .map(e -> e + 10)
//                .subscribeOn(Schedulers.single()) // 改变订阅者所在线程池
                .subscribe();
    }

    public static void thread1(String[] args) {
        Scheduler s = Schedulers.newParallel("p-s", 4);
        final Flux<String> flux = Flux
                .range(1, 2)
                .map(i -> 10 + i) // 没指定线程池，默认使用发布者的线程
                .log()
                .publishOn(s) // 该方法之后才会使用p-s线程池
                .map(i -> "value" + i)
                .log();
        // 只要不指定线程池，默认发布者用的线程就是订阅者的线程
        new Thread(() -> flux.subscribe(System.out::println), "thread1").start();
    }
}
